package com.rem.rabbitmq.ee.Fconfirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rem.rabbitmq.ee.RabbitMqUtils;

import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;


/**
 * 发布确认
 * 1.要求队列必须持久化
 * 2.要求队列里的消息必须持久化
 * 3.发布确认
 * <p>
 * <p>
 * 发布确认原理
 * 所有在该信道上面发布的消息都将会被指派一个唯一的ID，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，
 * 这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出
 * <p>
 * <p>
 * 异步确认发布  是通过函数回调来保证是否投递成功 可靠和效率都最高，实现比较复杂  性价比最高
 * confirm模式最大的好处在于他是异步的
 * 当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息
 *
 * @author Rem
 * @date 2021-12-24
 */

public class AsyncConfirmProducer {

    private final static int MESSAGE_COUNT = 1005;

    public static void main(String[] args) throws Exception {

        try (Channel channel = RabbitMqUtils.getChannel();) {
            //声明队列
            channel.queueDeclare(RabbitMqUtils.QUEUE_CONFIRM, true, false, false, null);
            /**
             * 启用发布确认
             */
            channel.confirmSelect();
            System.err.println("开始发送消息...");
            /**
             * 线程安全有序的哈希表，适用于高并发
             *  1.轻松的将序列号与消息关联
             *  2.轻松批量删除条目 只要给到序号
             *  3.支持高并发
             *
             */
            ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
            /**
             * 发布确认回调  消息确认成功
             * deliveryTag 消息的标记
             * multiple 是否为批量
             */
            ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
                ConcurrentNavigableMap<Long, String> confirmed = null;
                if (multiple) {
                    //返回的是小于等于当前序列号的确认消息 是一个map
                    confirmed = outstandingConfirms.headMap(deliveryTag);
                } else {
                    outstandingConfirms.remove(deliveryTag);
                }
                if (confirmed != null) {
                    confirmed.clear();
                }
                System.out.println("成功的消息" + deliveryTag);
            };
            /**
             * 发布确认回调 消息确认失败
             */
            ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
                System.out.println("未确认的编号" + deliveryTag + "消息:" + outstandingConfirms.get(deliveryTag));
            };
            /**
             * 消息监听器
             */
            channel.addConfirmListener(ackCallback, nackCallback);

            long begin = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                String message = i + ":confirm";

                channel.basicPublish("", RabbitMqUtils.QUEUE_CONFIRM, null, message.getBytes());
                outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            }
            long end = System.currentTimeMillis();
            System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (end - begin) + "ms");
        } finally {
            RabbitMqUtils.close();
        }

    }
}
